/*
 * (c) Copyright 2009 Hewlett-Packard Development Company, LP
 * All rights reserved.
 * [See end of file]
 */

package com.hp.hpl.jena.clusteredtdb;

import java.io.*;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.Iterator;

import com.hp.hpl.jena.clusteredtdb.base.UtilityFunctions;
import com.hp.hpl.jena.clusteredtdb.comms.ClusterRepresentation;
import com.hp.hpl.jena.clusteredtdb.comms.LocalVNode;
import com.hp.hpl.jena.clusteredtdb.comms.RemoteVNode;


/*
 * Basic TCP server to receive requests 
 * for subqueries, get the results, and then 
 * send them back.  Starting out with a single 
 * threaded example, then moving on to multi threaded/NIO
 * instead.
 */
public class DataNodeServer {
	final int DEBUG = 1;
	int threadsrunning;
	ServerSocket serverSock;
	HashMap<Integer,LocalVNode> vnodes;
	ClusteredTDBConfig ctdb;
	ClusterRepresentation cr;
	int serverId;
	int hits = 0;
	
	public DataNodeServer(String configFile) {
		ctdb = new ClusteredTDBConfig(configFile);
		cr = new ClusterRepresentation(ctdb);
		this.serverId = ctdb.getServerId();
		init(cr, serverId);
		listen();
	}
	
	protected void init(ClusterRepresentation cr, int serverId) {
		
		try {	
			
			serverSock = new ServerSocket(ctdb.getPort());
			serverSock.setReuseAddress(true);
			
		} catch (IOException e) {
			System.err.println("Unable to bind port "+ctdb.getPort()+": " +e);
			System.exit(-1);
		}
		
		HashMap<Integer,RemoteVNode> rvnodes = cr.getNodesForServer(serverId);
		if(rvnodes == null) {
			System.err.println("Unknown Server: "+serverId);
			System.exit(-1);
		}
		Iterator<Integer> it = rvnodes.keySet().iterator();
		vnodes = new HashMap<Integer,LocalVNode>();
		
		//Build the local representations of all the vnodes,
		while(it.hasNext()) {
			Integer i = it.next();
			vnodes.put(i, 
					new LocalVNode(i.intValue(),this, ctdb.getDataFileBase(), ctdb.mask));
		}
	}
	
	private void listen() {
		
		while(true) {
			
			Socket clientSock;
			try {
				clientSock = serverSock.accept();
				ServerConnection sc = new ServerConnection(clientSock,this);
				System.out.println("Listen - got connection") ;
				new Thread(sc).start();
				
		
			} catch (IOException e) {
				System.err.println("I/O error: " + e);
				e.printStackTrace();
			}
	
		}
		
	}

	public static void exec(String[] args) {
		String serverConfig = "resources/serverconfig.properties";
		if(args.length > 0) {
			serverConfig = args[0];
		}
		
		DataNodeServer dns = new DataNodeServer(serverConfig);
		
		
	}
	
}

class ServerConnection implements Runnable {

	DataNodeServer parent;
	Socket clientSock;
	PrintWriter out;
	BufferedReader in;
	
	public ServerConnection(Socket clientSocket, DataNodeServer parent) {
		this.clientSock = clientSocket;
		this.parent = parent;
		
		try {
			this.out = new PrintWriter(clientSock.getOutputStream());
			this.in = new BufferedReader(new InputStreamReader(clientSock.getInputStream()));
			
		} catch (IOException e) {
			System.err.println("I/O error: " + e);
		}
	}

	public void run() {
		String inputLine;
		boolean newreq = true;
		long time = System.currentTimeMillis();
		try {
			ArrayList<String> request = new ArrayList<String>();
			while ((inputLine = in.readLine()) != null) {	
				if(inputLine.equals("END")) {
					int reqSize = request.size();
					if(this.parent.DEBUG > 0) {
						System.out.println("Size: "+ reqSize);
					}
					if(reqSize > 1) {
						Collection<Integer> vnodeIds = this.getVNodes(request.get(0));
						request.remove(0);
						this.processInput(request,vnodeIds,out);
						//this.processInputTest(parent.connections,out);
					} else {
						out.println("ERROR");
						out.println("No request");
					}
					out.println("END");
					out.flush();
					parent.hits++;
					System.out.println("HITS: "+ parent.hits);
					if(this.parent.DEBUG > 0) {
						System.out.println("Finish: Time: "+ (System.currentTimeMillis() - time));
					}
					request = new ArrayList<String>();
					newreq = true;
				} else if(inputLine.equals("ENDCONNECTION") || clientSock.isClosed() == true) {
					clientSock.close();
					return;
				} else {
					if(newreq) {
						if(this.parent.DEBUG > 0) {
							time = System.currentTimeMillis();
							System.out.println("New: Time: "+ time);
						}
						newreq = false;
					}
					request.add(inputLine);
					if(this.parent.DEBUG > 1) {
						System.out.println("Server received: "+inputLine);
					}
				}
			}

			

		} catch (IOException e) {
			System.err.println("I/O error: " + e);
		}
	}

	protected Collection<Integer> getVNodes(String vnodeString) {
		int[] vnodeIds = UtilityFunctions.spacedStringToIntArray(vnodeString);
		Collection<Integer> integerVNodeIds;
		if(vnodeIds[0] == -1) {
			integerVNodeIds = parent.vnodes.keySet();
		} else {
			integerVNodeIds = new ArrayList<Integer>(vnodeIds.length);
			for(int i = 0; i < vnodeIds.length; i++) {
				integerVNodeIds.add(new Integer(vnodeIds[i]));
			}
		}
		return integerVNodeIds;
	}
	
	protected void processInput(ArrayList<String> request, Collection<Integer> vnodeIds, PrintWriter out) {
		
		Iterator<Integer> idIterator = vnodeIds.iterator();
		
		//TODO multi-thread this loop
		while (idIterator.hasNext()) {

			Iterator<String> reqIterator = request.iterator();
			LocalVNode lvn = parent.vnodes.get(idIterator.next());
			
			String requestType = reqIterator.next();
			
			//TODO turn these sizes into constants
			if (request.size() == 3 && requestType.equals("FIND")) {
				lvn.processFind(reqIterator,out);
			} else if (request.size() == 4 && requestType.equals("FINDRANGE")){
				lvn.processFindRange(reqIterator, out);
			} else if (request.size() > 1 && requestType.equals("IDTONODE")) {
				lvn.processIdToNode(reqIterator, out);
			} else if (requestType.equals("WRITENODE")) {
				lvn.processWriteNode(reqIterator, out);
			} else if (request.size() > 2 && requestType.equals("WRITERECORD")) {
				lvn.processWriteRecord(reqIterator, out);
			} else if (request.size() == 1 && requestType.equals("GETALLNODES")) {
				lvn.processGetAllNodes(out);
			} else if (request.size() == 1 && requestType.equals("CLOSEOBJECT")) {
				lvn.processCloseObject(out);
			} else if (request.size() == 2 && requestType.equals("SYNCOBJECT")) {
				lvn.processSyncObject(reqIterator, out);
			} else if (request.size() == 2 && requestType.equals("CLOSERINDEX")) {
				lvn.processCloseRIndex(reqIterator, out);
			} else if (request.size() == 3 && requestType.equals("SYNCRINDEX")) {
				lvn.processSyncRIndex(reqIterator, out);
			}
		}

	}
	
	
	
}

/*
 * (c) Copyright 2009 Hewlett-Packard Development Company, LP
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions
 * are met:
 * 1. Redistributions of source code must retain the above copyright
 *    notice, this list of conditions and the following disclaimer.
 * 2. Redistributions in binary form must reproduce the above copyright
 *    notice, this list of conditions and the following disclaimer in the
 *    documentation and/or other materials provided with the distribution.
 * 3. The name of the author may not be used to endorse or promote products
 *    derived from this software without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR
 * IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
 * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
 * IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT,
 * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT
 * NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
 * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
 * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF
 * THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 */
